package com.market.goods.comsumer;

import cn.hutool.json.JSONUtil;
import com.google.common.collect.Lists;
import com.market.goods.constance.TopicsEnum;
import com.market.goods.dao.es.CategoryBrandDocument;
import com.market.goods.dao.es.ESIndexNameEnum;
import com.market.goods.dao.es.MarketGoodsDocument;
import com.market.goods.dao.mongo.CategoryBrand;
import com.market.goods.dao.mongo.MarketGoods;
import com.market.goods.dao.mongo.MarketGoodsParams;
import com.market.goods.producer.RocketProducer;
import com.market.goods.service.MarketGoodsService;
import com.market.goods.service.MongoDbService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.List;

@Slf4j
@Component
public class KafkaClusterComsumer {

    @Autowired
    RocketProducer rocketProducer;

    @Autowired
    MongoDbService mongoDbService;

    @Autowired
    MarketGoodsService marketGoodsService;

    int j = 1;
    int a = 1;
    int b = 1;

    //    /**
//     * 第一种场景：监听一个Kafka 主题
//     * **/
//    @KafkaListener(topics = "send-topic")
////    public void receiveMessageFromKafka(ConsumerRecord<?, ?> consumer) {
//    public void receiveMessageFromKafka(String message) {
//        log.info("receiveMessageFromKafkaMessage={}",message);
////        System.out.println("topic名称:" + consumer.topic() + ",key:" + consumer.key() + ",分区位置:" + consumer.partition()
////                + ", 下标" + consumer.offset());
//
//    }
//
    List<Object> marketGoodsList = Lists.newArrayList();
    List<Object> marketGoodsDocumentList = Lists.newArrayList();
    @KafkaListener(topics = "marketGoods")
    public void sendToKafkaStandardMessageAsync(String message) {
//        log.info("j={}",j);
        MarketGoods marketGoods = JSONUtil.toBean(message, MarketGoods.class);
        MarketGoodsDocument darketGoodsDocument = JSONUtil.toBean(message, MarketGoodsDocument.class);
        j++;
        if (j >= 2000) {
            j = 1;
            //批量插入mogodb
            mongoDbService.insertAll(marketGoodsList);
            //批量插入mogodb
            marketGoodsService.inserBulktEs(ESIndexNameEnum.MARKET_GOODS.getIndexName(), marketGoodsDocumentList);
            marketGoodsList.clear();
            marketGoodsDocumentList.clear();
            log.info("sendToKafkaStandardMessageAsync");
        }else {
            marketGoodsList.add(marketGoods);
            marketGoodsDocumentList.add(darketGoodsDocument);
            return;
        }
    }

    @KafkaListener(topics = "marketGoods")
    public void sendToKafkaStandardMessageAsync3(String message) {
//        log.info("j={}",j);
        MarketGoods marketGoods = JSONUtil.toBean(message, MarketGoods.class);
        MarketGoodsDocument darketGoodsDocument = JSONUtil.toBean(message, MarketGoodsDocument.class);
        j++;
        if (j >= 2000) {
            j = 1;
            //批量插入mogodb
            mongoDbService.insertAll(marketGoodsList);
            //批量插入mogodb
            marketGoodsService.inserBulktEs(ESIndexNameEnum.MARKET_GOODS.getIndexName(), marketGoodsDocumentList);
            marketGoodsList.clear();
            marketGoodsDocumentList.clear();
            log.info("sendToKafkaStandardMessageAsync");
        }else {
            marketGoodsList.add(marketGoods);
            marketGoodsDocumentList.add(darketGoodsDocument);
            return;
        }
    }
        List<Object> categoryBrandList = Lists.newArrayList();
        List<Object> categoryBrandDocumentList = Lists.newArrayList();
    @KafkaListener(topics = "categoryBrand")
    public void categoryBrandAsync(String message) {
//        log.info("a={}",a);
        CategoryBrand categoryBrand = JSONUtil.toBean(message, CategoryBrand.class);
        CategoryBrandDocument categoryBrandDocument = JSONUtil.toBean(message, CategoryBrandDocument.class);
        a++;
        if (a >= 2000) {
            a = 1;
            //批量插入mogodb
            mongoDbService.insertAll(categoryBrandList);
            //批量插入mogodb
            marketGoodsService.inserBulktEs(ESIndexNameEnum.CATEGORY_BRAND.getIndexName(), categoryBrandDocumentList);
            categoryBrandList.clear();
            categoryBrandDocumentList.clear();
            log.info("categoryBrandAsync");
        }else {
            categoryBrandList.add(categoryBrand);
            categoryBrandDocumentList.add(categoryBrandDocument);
            return;
        }

    }

    @KafkaListener(topics = "categoryBrand")
    public void categoryBrandAsync2(String message) {
        log.info("a={}",a);
        CategoryBrand categoryBrand = JSONUtil.toBean(message, CategoryBrand.class);
        CategoryBrandDocument categoryBrandDocument = JSONUtil.toBean(message, CategoryBrandDocument.class);
        a++;
        if (a >= 2000) {
            a = 1;
            //批量插入mogodb
            mongoDbService.insertAll(categoryBrandList);
            //批量插入mogodb
            marketGoodsService.inserBulktEs(ESIndexNameEnum.CATEGORY_BRAND.getIndexName(), categoryBrandDocumentList);
            categoryBrandList.clear();
            categoryBrandDocumentList.clear();
            log.info("categoryBrandAsync");
        }else {
            categoryBrandList.add(categoryBrand);
            categoryBrandDocumentList.add(categoryBrandDocument);
            return;
        }

    }

    List<Object> marketGoodsParamsList = Lists.newArrayList();
    @KafkaListener(topics = "marketGoodsParams")
    public void marketGoodsParams1Async(String message) {
//        log.info("b={}",b);
        MarketGoodsParams marketGoodsParams = JSONUtil.toBean(message, MarketGoodsParams.class);
            b++;
        if (b >= 2000) {
            b = 1;
            //批量插入mogodb
            mongoDbService.insertAll(marketGoodsParamsList);
            marketGoodsParamsList.clear();
            log.info("marketGoodsParams1Async");
        }else {
            marketGoodsParamsList.add(marketGoodsParams);
            return;
        }

    }

    @KafkaListener(topics = "marketGoodsParams")
    public void marketGoodsParams2Async(String message) {
//        log.info("b={}",b);
        MarketGoodsParams marketGoodsParams = JSONUtil.toBean(message, MarketGoodsParams.class);
        b++;
        if (b >= 2000) {
            b = 1;
            //批量插入mogodb
            mongoDbService.insertAll(marketGoodsParamsList);
            marketGoodsParamsList.clear();
            log.info("marketGoodsParams1Async");
        }else {
            marketGoodsParamsList.add(marketGoodsParams);
            return;
        }

    }

//    @KafkaListener(topics = "marketGoodsParams")
//    public void marketGoodsParams2Async(String message) {
////        log.info("marketGoodsParams2AsyncMessage={}",message);
//        MarketGoodsParams marketGoodsParams = JSONUtil.toBean(message, MarketGoodsParams.class);
//        //插入mogodb
//        mongoDbService.insertServer(marketGoodsParams);
//        //插入clickhouse
//
//
//    }
//    @KafkaListener(topics = "sync-topic")
////    public void receiveMessageFromKafka(ConsumerRecord<?, ?> consumer) {
//    public void sendToKafkaStandardMessageSync(String message) {
//        log.info("sendToKafkaStandardMessageSyncMessage={}",message);
////        System.out.println("topic名称:" + consumer.topic() + ",key:" + consumer.key() + ",分区位置:" + consumer.partition()
////                + ", 下标" + consumer.offset());
//
//    }
//
//    @KafkaListener(topics = "transaction-topic")
////    public void receiveMessageFromKafka(ConsumerRecord<?, ?> consumer) {
//    public void executeInTransaction(String message) {
//        log.info("executeInTransactionMessage={}",message);
////        System.out.println("topic名称:" + consumer.topic() + ",key:" + consumer.key() + ",分区位置:" + consumer.partition()
////                + ", 下标" + consumer.offset());
//
//    }

//    @KafkaListener(topics = "markets")
////    public void receiveMessageFromKafka(ConsumerRecord<?, ?> consumer) {
//    public void getMarketTables1(String message) {
//        log.info("getMarketTables={}", message);
////        System.out.println("topic名称:" + consumer.topic() + ",key:" + consumer.key() + ",分区位置:" + consumer.partition()
////                + ", 下标" + consumer.offset());
//
//    }
//
//    @KafkaListener(topics = "markets")
////    public void receiveMessageFromKafka(ConsumerRecord<?, ?> consumer) {
//    public void getMarketTables2(String message) {
//        log.info("getMarketTables={}", message);
////        System.out.println("topic名称:" + consumer.topic() + ",key:" + consumer.key() + ",分区位置:" + consumer.partition()
////                + ", 下标" + consumer.offset());
//
//    }
//
//    @KafkaListener(topics = "markets")
////    public void receiveMessageFromKafka(ConsumerRecord<?, ?> consumer) {
//    public void getMarketTables3(String message) {
//        log.info("getMarketTables={}", message);
////        System.out.println("topic名称:" + consumer.topic() + ",key:" + consumer.key() + ",分区位置:" + consumer.partition()
////                + ", 下标" + consumer.offset());
//
//    }
//
//    @KafkaListener(topics = "markets")
////    public void receiveMessageFromKafka(ConsumerRecord<?, ?> consumer) {
//    public void getMarketTables4(String message) {
//        log.info("getMarketTables={}", message);
////        System.out.println("topic名称:" + consumer.topic() + ",key:" + consumer.key() + ",分区位置:" + consumer.partition()
////                + ", 下标" + consumer.offset());
//
//    }
}
